package com.atguigu.edu.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.app.func.DimAsyncFunction;
import com.atguigu.edu.realtime.bean.PaperCourseTestBean;
import com.atguigu.edu.realtime.bean.PaperScoreSectionBean;
import com.atguigu.edu.realtime.util.DateFormatUtil;
import com.atguigu.edu.realtime.util.MyClickhouseUtil;
import com.atguigu.edu.realtime.util.MyKafkaUtil;
import com.atguigu.edu.realtime.util.TimestampLtz3CompareUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * @Author zhangsan
 * @Date 2022/10/20 14:02
 * @Description //TODO dws层 各试卷分数段
 */
public class DwsTestScoreSectionWindow {
    public static void main(String[] args) throws Exception{
        //TODO 1.基本环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        //TODO 2.检查点设置
        //TODO 3.从kafka主题中读取数据 dwd_test_paper_course_test
        String topic = "dwd_test_paper_course_test";
        String groupId = "dws_test_score_section_window";
        FlinkKafkaConsumer<String> kafkaConsumer = MyKafkaUtil.getKafkaConsumer(topic, groupId);
        DataStreamSource<String> kafkaStrDS = env.addSource(kafkaConsumer);
        //TODO 4.对读取的数据进行类型转换 jsonStr-> jsonObj
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.map(JSON::parseObject);
        //TODO 5.按照id进行 分组
        KeyedStream<JSONObject, String> idKeyedDS = jsonObjDS.keyBy(jsonObject -> jsonObject.getString("id"));
        //TODO 6.使用flink状态编程 + 定时器去重
        SingleOutputStreamOperator<JSONObject> distinctDS = idKeyedDS.process(
                new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                    private ValueState<JSONObject> lastJsonObjState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        getRuntimeContext().getState(new ValueStateDescriptor<JSONObject>("lastJsonObjState", JSONObject.class));
                    }

                    @Override
                    public void processElement(JSONObject jsonObj, Context ctx, Collector<JSONObject> out) throws Exception {
                        JSONObject lastJsonObj = lastJsonObjState.value();
                        if (lastJsonObj == null) {
                            lastJsonObjState.update(jsonObj);
                            long currentProcessingTime = ctx.timerService().currentProcessingTime();
                            ctx.timerService().registerProcessingTimeTimer(currentProcessingTime + 5000L);
                        } else {
                            String lastTs = lastJsonObj.getString("ts");
                            String curTs = jsonObj.getString("ts");
                            if (TimestampLtz3CompareUtil.compare(lastTs, curTs) <= 0) {
                                lastJsonObjState.update(jsonObj);
                            }
                        }
                    }

                    @Override
                    public void onTimer(long timestamp, OnTimerContext ctx, Collector<JSONObject> out) throws Exception {
                        JSONObject jsonObj = lastJsonObjState.value();
                        if (jsonObj != null) {
                            out.collect(jsonObj);
                        }
                        lastJsonObjState.clear();
                    }
                }
        );
        //TODO 7. 类型转换 封装为实体类
        SingleOutputStreamOperator<PaperScoreSectionBean> scoreBean = distinctDS.map(
                new MapFunction<JSONObject, PaperScoreSectionBean>() {
                    @Override
                    public PaperScoreSectionBean map(JSONObject jsonObj) throws Exception {
                        String paperId = jsonObj.getString("paper_id");
                        Double score = Double.valueOf(jsonObj.getString("score"));
                        Long ts = jsonObj.getLong("ts");
                        String scoreSection = null;

                        if (score < 60) {
                            scoreSection = "0-59";
                        } else if (score >= 60 && score < 70) {
                            scoreSection = "60-69";
                        } else if (score >= 70 && score < 80) {
                            scoreSection = "70-79";
                        } else if (score >= 80 && score < 90) {
                            scoreSection = "80-89";
                        } else if (score >= 90 && score <= 100) {
                            scoreSection = "90-100";
                        }
                        PaperScoreSectionBean scoreBean = PaperScoreSectionBean.builder()
                                .paperId(paperId)
                                .score(score)
                                .scoreSection(scoreSection)
                                .userCnt(1L)
                                .ts(ts)
                                .build();

                        return scoreBean;
                    }
                }
        );
        //TODO 8.指定watermark以及提取事件时间字段
        SingleOutputStreamOperator<PaperScoreSectionBean> withWatermarkDS = scoreBean.assignTimestampsAndWatermarks(
                WatermarkStrategy.<PaperScoreSectionBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<PaperScoreSectionBean>() {
                                    @Override
                                    public long extractTimestamp(PaperScoreSectionBean scoreBean, long l) {
                                        return scoreBean.getTs();
                                    }
                                }
                        )
        );
        //TODO 9.按照试卷id进行分组  在根据分数段分组
        KeyedStream<PaperScoreSectionBean, String> paperIdKeyedDS = withWatermarkDS.keyBy(PaperScoreSectionBean::getPaperId);
        KeyedStream<PaperScoreSectionBean, String> scoreKeyedDS = paperIdKeyedDS.keyBy(PaperScoreSectionBean::getScoreSection);
        //TODO 10.开窗
        WindowedStream<PaperScoreSectionBean, String, TimeWindow> windowDS = scoreKeyedDS.window(TumblingEventTimeWindows.of(Time.seconds(10)));
        //TODO 11.聚合
        SingleOutputStreamOperator<PaperScoreSectionBean> reduceDS = windowDS.reduce(
                new ReduceFunction<PaperScoreSectionBean>() {
                    @Override
                    public PaperScoreSectionBean reduce(PaperScoreSectionBean value1, PaperScoreSectionBean value2) throws Exception {
                        value1.setUserCnt(value1.getUserCnt() + value2.getUserCnt());
                        return value1;
                    }
                },
                new WindowFunction<PaperScoreSectionBean, PaperScoreSectionBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window, Iterable<PaperScoreSectionBean> input, Collector<PaperScoreSectionBean> out) throws Exception {
                        for (PaperScoreSectionBean scoreBean : input) {
                            scoreBean.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                            scoreBean.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                            scoreBean.setTs(System.currentTimeMillis());
                            out.collect(scoreBean);
                        }

                    }
                }
        );
        //TODO 12.和维度表  dim_test_paper  关联
        SingleOutputStreamOperator<PaperScoreSectionBean> withPaperDS = AsyncDataStream.unorderedWait(
                reduceDS,
                new DimAsyncFunction<PaperScoreSectionBean>("dim_test_paper") {
                    @Override
                    public void join(PaperScoreSectionBean scoreBean, JSONObject dimInfoJsonObj) {
                        scoreBean.setPaperName(dimInfoJsonObj.getString("PAPER_TITLE"));
                    }
                    @Override
                    public String getKey(PaperScoreSectionBean scoreBean) {
                        return scoreBean.getPaperId();
                    }
                }, 60, TimeUnit.SECONDS
        );
        //TODO 13.将结果写入到ck中
        withPaperDS.print(">>>>>");
        withPaperDS.addSink(MyClickhouseUtil.getSinkFunction("insert into dws_test_score_section_window values(?,?,?,?,?,?,?)"));
        env.execute();

    }
}
